package com.easydata.agent.core.dataIO;

import com.easydata.agent.config.TaskConfiguration;
import com.easydata.agent.core.DAGContext;
import com.easydata.agent.exceptions.TaskException;
import com.easydata.libPlugin.core.IDataWriter;

import java.io.File;
import java.io.FileOutputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

public class FileCacheDataWriter implements IDataWriter {
    private DAGContext context;
    private List<String> destination = new ArrayList<>();
    private List<FileOutputStream> foss = new ArrayList<>();

    public FileCacheDataWriter(DAGContext context) {
        this.context = context;
    }

    public void addDestination(String id) {
        destination.add(id);
    }

    @Override
    public void open() throws Exception {
        String path = TaskConfiguration.getInstance().getCacheDir() + "/" + this.context.getTaskId() + "/";
        var file = new File(path);

        if (!file.exists()) {
            if (!file.mkdirs()) {
                throw new TaskException(-1, "can not create the cache dir");
            }
        }

        for (String id : destination) {
            var fos = new FileOutputStream(path + id + ".bin", true);
            foss.add(fos);
        }
    }

    @Override
    public void write(byte[] obj) throws Exception {

        var lenBuffer = ByteBuffer.allocate(4);
        lenBuffer.putInt(obj.length);

        for (var fos : foss) {
            fos.write(lenBuffer.array());
            fos.write(obj);
        }
    }

    @Override
    public int size() throws Exception {
        throw new TaskException(-1, "not supported");
    }

    @Override
    public void close() throws Exception {
        for (var fos : foss) {
            fos.close();
        }
    }
}
